package com.sbp.message.container;

import com.sbp.message.api.common.CommonUtils;
import com.sbp.message.api.common.TopicUtils;
import com.sbp.message.api.service.MessageService;
import com.sbp.message.conf.ActiveMQConfig;
import com.sbp.message.proxy.MessageInvoker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.util.StringUtils;

import javax.jms.ConnectionFactory;
import javax.jms.MessageListener;
import javax.jms.Session;
import java.util.ArrayList;

public class MessageListenerContainer implements InitializingBean, ApplicationContextAware {
    private static final Logger logger = LoggerFactory.getLogger(MessageListenerContainer.class);

    private static ConfigurableApplicationContext applicationContext; // Spring应用上下文环境

    private ActiveMQConfig config;
    public void setConfig(ActiveMQConfig config) {
        this.config = config;
    }

    private MessageService messageService;
    public void setMessageService(MessageService messageService) {
        this.messageService = messageService;
    }

    @Override
    public void afterPropertiesSet() {
        registerQueuesListener();
        registerTopicsListener();
    }


    private void registerOneQueueListener(DefaultListableBeanFactory dbf, ConnectionFactory connectionFactory, String destination, boolean consumerExclusive) {
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        container.setConcurrentConsumers(1);
        container.setConnectionFactory(connectionFactory);
        container.setDestinationName(destination+"?consumer.exclusive="+consumerExclusive);
        container.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
        container.setSessionTransacted(false);
        container.setMessageListener(MessageInvoker.getInstance(MessageListener.class));

        container.afterPropertiesSet();
        container.start();
        dbf.registerSingleton(destination + "ListenerContainer", container);
    }
    private void registerQueuesListener() {
        // 获取queue数组
        if (StringUtils.isEmpty(config.getAmqQueueDestination()))
            return;
        ArrayList<String> destinationNameList = CommonUtils.splitByComma(config.getAmqQueueDestination());
        ArrayList<String> consumerExclusiveStrList = CommonUtils.splitByComma(config.getAmqQueueConsumerExclusive());

        DefaultListableBeanFactory dbf = (DefaultListableBeanFactory) applicationContext.getBeanFactory();
        ConnectionFactory connectionFactory = (ConnectionFactory) dbf.getBean("connectionFactory");

        // 创建队列监听容器
        for (int i=0; i<destinationNameList.size(); ++i) {
            String destination = destinationNameList.get(i);
            boolean consumerExclusive = Boolean.parseBoolean(consumerExclusiveStrList.get(i));
            try {
                registerOneQueueListener(dbf, connectionFactory, destination, consumerExclusive);
            } catch (Exception e) {
                logger.warn("", e);
                throw e;
            }
        }
    }


    private String registerOneTopicListener(DefaultListableBeanFactory dbf, ConnectionFactory connectionFactory, String topicSuffix, boolean consumerExclusive) {
        // 为了防止非法字符, 去除了appName中的非法字符
        String consumersDescribe = TopicUtils.generateConsumersDescribe(config.getAppName(), topicSuffix);

        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        container.setConcurrentConsumers(1);
        container.setConnectionFactory(connectionFactory);
        container.setDestinationName(consumersDescribe+"?consumer.exclusive="+consumerExclusive);
        container.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
        container.setSessionTransacted(false);
        container.setMessageListener(MessageInvoker.getInstance(MessageListener.class));

        container.afterPropertiesSet();
        container.start();
        dbf.registerSingleton(consumersDescribe + "ListenerContainer", container);

        return consumersDescribe;
    }
    private void registerTopicsListener() {
        // 获取虚拟主题数组
        if (StringUtils.isEmpty(config.getAmqTopicDestinationSuffix()))
            return;
        ArrayList<String> destinationSuffixList = CommonUtils.splitByComma(config.getAmqTopicDestinationSuffix());
        ArrayList<String> consumerExclusiveStrList = CommonUtils.splitByComma(config.getAmqTopicConsumerExclusive());

        DefaultListableBeanFactory dbf = (DefaultListableBeanFactory) applicationContext.getBeanFactory();
        ConnectionFactory connectionFactory = (ConnectionFactory) dbf.getBean("connectionFactory");

        // 创建主题监听容器
        for (int i=0; i<destinationSuffixList.size(); ++i) {
            String topicSuffix = destinationSuffixList.get(i);
            boolean consumerExclusive = Boolean.parseBoolean(consumerExclusiveStrList.get(i));
            try {
                String consumersDescribe = registerOneTopicListener(dbf, connectionFactory, topicSuffix, consumerExclusive);

                // 向消息中心注册消费者数量,用于一对多虚拟分组
                messageService.updateTopicRegistryByConsumersDescribe(consumersDescribe);
            } catch (Exception e) {
                logger.warn("", e);
                throw e;
            }
        }
    }



    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }

}
